package com.zy.rpc.provider.services;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;


/**
 * @author zy
 * @date 2023/8/29 21:01
 */
@Component
public class HotStorageOptimizer {

    /**
     * 存储策略
     * 所有用户的文件都会被存储在user/ 下
     * 而用户目录下有两个文件夹 hot和cold
     * 举例 user/zhangsan/hot/1.txt   user/zhangsan/cold/2.txt
     */

    /**
     * 遍历所有用户的 hot 文件夹，获取所有用户的 hot 文件夹路径。
     *
     * 对于每个用户，遍历其 hot 文件夹中的文件，获取每个文件的访问时间戳。
     *
     * 比较每个文件的访问时间戳与当前时间，计算距离上次访问的天数。
     *
     * 如果某个文件的访问天数超过三天，标记该文件。
     *
     * 在下一轮的处理中，检查标记的文件，如果该文件又被标记，将其移动到对应用户的 cold 文件夹中。
     *
     * 通过此存储策略用以模拟将访问频率不同的文件存储在速度不同的存储介质上（不同的存储介质用hot和cold目录模拟）
     */

    /**
     * 拿到所有的hot文件夹
     */
    @Scheduled(cron = "0 0 0 */3 * ?") // 每三天执行一次，每天的 00:00:00
    private  void hot() {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://node01:8020");

        try {
            FileSystem fs = FileSystem.get(conf);
            Path userRoot = new Path("/user");
            FileStatus[] userFolders = fs.listStatus(userRoot);
            for (FileStatus userFolder : userFolders) {
                if (userFolder.isDirectory()) {
                    processUserHotFolder(fs, userFolder.getPath());
                }
            }
            fs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private  void processUserHotFolder(FileSystem fs, Path userHotFolder) throws IOException {
        //拿到用户hot目录下所有的文件
        FileStatus[] hotFiles = fs.listStatus(userHotFolder);
        for (FileStatus fileStatus : hotFiles) {
            long accessTimeMillis = fileStatus.getAccessTime();
            long currentTimeMillis = System.currentTimeMillis();
            long dayInMillis = 24 * 60 * 60 * 1000;
            int daysPassed = (int) ((currentTimeMillis - accessTimeMillis) / dayInMillis);
            //如果三天没有被访问，则标记
            if (daysPassed > 3) {
                String markerPath = fileStatus.getPath().toString() + ".migrationMarker";
                if (!fs.exists(new Path(markerPath))) {
                    markFileForMigration(fs, fileStatus.getPath());
                }else {
                    migrateFilesToCold(fs, fileStatus);
                }
            }
        }
    }

    /**
     * 标记文件
     * @param fs
     * @param filePath
     * @throws IOException
     */
    private  void markFileForMigration(FileSystem fs, Path filePath) throws IOException {
        String markerPath = filePath.toString() + ".migrationMarker";
        fs.createNewFile(new Path(markerPath));
    }

    /**
     * 迁移文件
     * @param fs
     * @param fileStatus
     * @throws IOException
     */
    @Async("asyncExecutor")
    public void migrateFilesToCold(FileSystem fs, FileStatus fileStatus) throws IOException {
        String filePath = fileStatus.getPath().toString();
        //删除标记文件
        String markerPath = filePath + ".migrationMarker";
        fs.delete(new Path(markerPath), false);
        String coldFilePath = filePath.replace("/hot/", "/cold/");
        fs.rename(new Path(filePath), new Path(coldFilePath));
    }
}
